package cn.jstars.datatocloud.modules.base;

import com.rabbitmq.client.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.Connection;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * rabbitmq服务类
 *
 * @author Elvish
 */
@Component
public class MqService {
    /**
     * logback logger object
     **/
    private static final Logger LOGGER = LoggerFactory.getLogger(MqService.class);
    /**
     * sync listener thread pool size
     **/
    private static final Integer LISTENER_POOL_SIZE = 128;
    /**
     * Max Concurrent Consumers size
     **/
    private static final Integer MAX_CONCURRENT_CONSUMERS = 200;
    /**
     * Concurrent Consumers size
     **/
    private static final Integer CONCURRENT_CONSUMERS = 4;

    private static final long RETRY_DECLARATION_INTERVAL = 30 * 1000L;

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Autowired
    private ConnectionFactory connectionFactory;

    /**
     * send message to queue
     *
     * @param queue   queue name
     * @param content request content
     * @throws Exception exception
     */
    public void convertAndSend(final String queue, final String content) throws Exception {
        MessageProperties propertis = new MessageProperties();
        propertis.setDeliveryMode(MessageDeliveryMode.PERSISTENT);
        Message message = new Message(content.getBytes("UTF-8"), propertis);

        RabbitTemplate template  = getRabbitTemplate(connectionFactory);
        template.setConfirmCallback((correlationData, ack, cause) -> {
            if (!ack) {
                // 发送失败重新发送
                reConvertAndSend(queue, content);
            }
        });
        template.convertAndSend("", queue, message);
    }

    /**
     * resend message to queue
     *
     * @param queue   queue name
     * @param content request content
     */
    private void reConvertAndSend(final String queue, final String content) {
        try {
            LOGGER.error("发送失败消息为 ： {}", content);
            Thread.sleep(60000);
            convertAndSend(queue, content);
        } catch (Exception e) {
            LOGGER.error("发送失败", e);
        }

    }

    /**
     * 发送rpc请求
     *
     * @param queue      队列
     * @param requestMsg 请求消息
     * @return 响应消息
     */
    public String convertSendAndReceive(String queue, String requestMsg) {
        Object obj = rabbitTemplate.convertSendAndReceive("", queue, requestMsg);

        if (obj == null) {
            return null;
        }

        if (obj instanceof byte[]) {
            byte[] bytes = (byte[]) obj;
            return new String(bytes);
        }

        return obj.toString();
    }

    /**
     * get mq connection
     *
     * @return mq connection
     */
    private Connection getMqConn() {
        return connectionFactory.createConnection();
    }

    /**
     * create Channel client
     *
     * @return Channel client
     */
    private Channel createChannel() {
        return getMqConn().createChannel(false);
    }

    /**
     * create queue
     *
     * @param queue queue name
     */
    private void createQueue(String queue) {
        try {
            createChannel().queueDeclare(queue, true, false, false, null);
        } catch (IOException e) {
            LOGGER.error("create queue error!", e);
        }
    }

    /**
     * get queue message count
     *
     * @param queueName queueName
     * @return queue message count
     */
    public long getQueueMessageCount(String queueName) {
        try {
            return createChannel().messageCount(queueName);
        } catch (IOException e) {
            LOGGER.error("get queue message count error!", e);
        }
        return 0;
    }

    /**
     * queueBind
     *
     * @param queue      queue
     * @param exchange   exchange
     * @param routingKey routingKey
     */
    public void queueBind(String queue, String exchange, String routingKey) {
        try {
            createChannel().queueBind(queue, exchange, routingKey);
        } catch (IOException e) {
            LOGGER.error("queue bind error!", e);
        }
    }

    /**
     * queueUnbind
     *
     * @param queue      queue
     * @param exchange   exchange
     * @param routingKey routingKey
     */
    public void queueUnbind(String queue, String exchange, String routingKey) {
        try {
            createChannel().queueUnbind(queue, exchange, routingKey);
        } catch (IOException e) {
            LOGGER.error("queue Unbind error!", e);
        }
    }

    /**
     * get SimpleMessageListenerContainer
     *
     * @param queue queue name
     * @return SimpleMessageListenerContainer object
     */
    public SimpleMessageListenerContainer getContainer(String queue) {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
        container.setQueues(new Queue(queue));
        container.setExposeListenerChannel(true);
        container.setMaxConcurrentConsumers(MAX_CONCURRENT_CONSUMERS);
        container.setConcurrentConsumers(CONCURRENT_CONSUMERS);
        ExecutorService executorService = Executors.newFixedThreadPool(LISTENER_POOL_SIZE);
        container.setTaskExecutor(executorService);
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        container.setDeclarationRetries(Integer.MAX_VALUE);
        container.setRetryDeclarationInterval(RETRY_DECLARATION_INTERVAL);
        return container;
    }

    /**
     * 获取RabbitTemplate操作类
     * @param connectionFactory mq连接工厂
     * @return RabbitTemplate
     */
    public RabbitTemplate getRabbitTemplate(ConnectionFactory connectionFactory){
        RabbitTemplate template = new RabbitTemplate(connectionFactory);
        template.setMessageConverter(new Jackson2JsonMessageConverter());
        template.setMandatory(true);
        return template;
    }
}
